package com.xxx.work;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * Rabbit MQ使用场景二：work模式，1个生产者，多个消费者
 * 消费者轮流（round-robin）消费队列里的消息，且1条消息只能被1个消费者消费。
 * 
 * @author guweiqiang
 * 2018年6月13日
 */
public class ConsumerWorker2 {
	
	private static final String QUEUE_NAME = "work_queue";
//	private static final String HOST = "localhost";
	
	public static void main(String[] args) throws IOException, TimeoutException {
		// 1.创建连接工厂，并设置工厂相关属性
		ConnectionFactory factory = new ConnectionFactory();
//		factory.setHost(HOST); // 默认为localhost
		
		// 2.创建一个连接
		Connection connection = factory.newConnection();
		
		// 3.创建一个通道
		Channel channel = connection.createChannel();
		
		// 4.声明一个队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		channel.basicQos(1); // 每次从队列里获取1条消息进行消费（该参数只有当不同的消费者的处理速度不一致时才会起作用，和ack配合使用）
		/**
		 * 使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。
		 * 当消息处理完毕后，有了反馈，才会进行第二次发送。还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。
		 */
		
		// 5.从队列里接收消息
		System.out.println("ConsumerWorker2 Waiting for messages：");
		// DefaultConsumer表示consumer采取的是pull模式来消费消息
		Consumer consumer = new DefaultConsumer(channel) {
			// 重写handleDelivery方法
			@Override
			public void handleDelivery(String consumerTag,
					                    Envelope envelope,
					                    AMQP.BasicProperties properties,
					                    byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
                System.out.println("ConsumerWorker2 received message：" + message);
				try {
//					throw new Exception();
					dowork();
				} catch (Exception e) {
					e.printStackTrace();
					channel.abort();
				} finally {
					System.out.println("Worker2 done");
					channel.basicAck(envelope.getDeliveryTag(), false); // false表示使用手动确认模式
				}
			}
		};
		
		// 6.消费消息之后回复生产者（批量消费消息后1次性回复）
		channel.basicConsume(QUEUE_NAME, false, consumer); // false表示不自动回复（手动回复）
	}
	
	// 让线程睡眠一段时间，模拟要完成的业务
	private static void dowork() throws InterruptedException {
		Thread.sleep(1000);	// 模拟worker2消费消息时处理的时间比较短（因此被分配到的消息数量就多）
	}

}
